package org.springframework.boot.netty.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.netty.NettyProperties;
import org.springframework.boot.autoconfigure.netty.enums.StartUpMode;
import org.springframework.boot.netty.annotation.NettySocketListener;
import org.springframework.boot.netty.argumentResolver.*;
import org.springframework.boot.netty.exception.ConsumerStartupException;
import org.springframework.boot.netty.filter.JwtSecurityFilter;
import org.springframework.boot.netty.sync.AsyncMessageProcessing;
import org.springframework.boot.netty.sync.MessageBlockQueue;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.annotation.MergedAnnotations;
import org.springframework.core.convert.ConversionService;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.AnnotatedElement;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;


/**
 * Author: huoxingzhi
 * Date: 2020/12/10
 * Email: hxz_798561819@163.com
 */

/**
 * 作为监听器，监听器channel连接时间，启动对应线程
 */
public class NettyMessageListener implements ApplicationListener<ApplicationEvent>, ApplicationContextAware, BeanPostProcessor, SmartInitializingSingleton ,DisposableBean{


    private ApplicationContext applicationContext;

    private NettyProperties nettyProperties;

    private Object target;

    private Set<AsyncMessageProcessing> asyncMessageProcessings = new HashSet<>();

    private ConversionService conversionService;

    private HandlerMethodArgumentResolverComposite handlerMethodArgumentResolverComposite;

    private ReturnValueArgumentResolverComposite returnValueArgumentResolverComposite;

    private Map<Integer,Boolean> threadStartFlag = new ConcurrentHashMap<>(16);

    Executor threadPoolTaskExecutor;

    private JwtSecurityFilter jwtSecurityFilter;

    private ThrowableResolverComposite throwableResolverComposite;

    private final static Logger logger = LoggerFactory.getLogger(NettyMessageListener.class);


    @Override
    public void onApplicationEvent(ApplicationEvent event) {
        if(event instanceof ContextRefreshedEvent) {
            ContextRefreshedEvent contextRefreshedEvent = (ContextRefreshedEvent) event;
            if (contextRefreshedEvent.getApplicationContext().getParent() == null) {
                this.initializeConsumers();
                waitForConsumersToStart(this.asyncMessageProcessings);
            }
        }

        if(event instanceof SocketEvent){

        }
    }



    protected int initializeConsumers() {
        // 启动线程
        for (int mapIndex = 0; mapIndex < this.nettyProperties.getMessageMapCapacity(); mapIndex++) {
            // 双重if防止多线程
            if (threadStartFlag.get(mapIndex) == Boolean.FALSE) {
                synchronized (this) {
                    if (threadStartFlag.get(mapIndex) == Boolean.FALSE) {
                        AsyncMessageProcessing asyncMessageProcessing = new AsyncMessageProcessing(
                                this.nettyProperties,
                                this.target,
                                this.conversionService,
                                this.handlerMethodArgumentResolverComposite,
                                this.returnValueArgumentResolverComposite,
                                this.jwtSecurityFilter,
                                this.throwableResolverComposite,
                                mapIndex
                        );
                        this.asyncMessageProcessings.add(asyncMessageProcessing);
                        this.threadStartFlag.remove(mapIndex);
                        this.threadStartFlag.put(mapIndex, Boolean.TRUE);
                        this.threadPoolTaskExecutor.execute(asyncMessageProcessing);
                        this.asyncMessageProcessings.add(asyncMessageProcessing);
                    }
                }
            }
        }
        return this.asyncMessageProcessings.size();
    }

    private void waitForConsumersToStart(Set<AsyncMessageProcessing> processors) {
        for (AsyncMessageProcessing processor : processors) {
            ConsumerStartupException startupException = null;
            try {
                startupException = processor.getStartupException();
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new ConsumerStartupException("Fatal exception on listener startup");
            }
            if (startupException != null) {
                throw new ConsumerStartupException("Fatal exception on listener startup");
            }
        }
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

        // find targetClass class from an AOP proxy
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        Collection<NettySocketListener> listenerAnnotations = findListenerAnnotations(targetClass);
        if(!CollectionUtils.isEmpty(listenerAnnotations)){
            this.target = bean;
        }
        return bean;
    }

    private Collection<NettySocketListener> findListenerAnnotations(AnnotatedElement element) {
        return MergedAnnotations.from(element, MergedAnnotations.SearchStrategy.TYPE_HIERARCHY)
                .stream(NettySocketListener.class)
                .map(ann -> ann.synthesize())
                .collect(Collectors.toList());
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterSingletonsInstantiated() {

        this.nettyProperties = this.applicationContext.getBean(NettyProperties.class);

        Map<String, HandlerMethodArgumentResolver> handlerMethodArgumentResolverMap = this.applicationContext.getBeansOfType(HandlerMethodArgumentResolver.class);
        this.handlerMethodArgumentResolverComposite = new HandlerMethodArgumentResolverComposite(handlerMethodArgumentResolverMap);

        List<ReturnValueArgumentResolver> resolverComposites = this.applicationContext.getBeansOfType(ReturnValueArgumentResolver.class).values().stream().collect(Collectors.toList());
        this.returnValueArgumentResolverComposite = new ReturnValueArgumentResolverComposite(resolverComposites);


        this.threadPoolTaskExecutor = (Executor) this.applicationContext.getBean( "process-executor");
        try {
            this.conversionService = this.applicationContext.getBean(ConversionService.class);
        }catch (BeansException e){
            logger.debug("IOC container not found bean of conversionService.class");
        }

        if(this.applicationContext.containsBean("shiroConfig")){
            this.jwtSecurityFilter = this.applicationContext.getBean(JwtSecurityFilter.class);
        }

        List<ThrowableResolver> throwableResolvers = this.applicationContext.getBeansOfType(ThrowableResolver.class).values().stream().collect(Collectors.toList());
        this.throwableResolverComposite = new ThrowableResolverComposite(throwableResolvers);

        for (int init = 0; init <this.nettyProperties.getMessageMapCapacity(); init++) {
            //初始化线程启动标记
            threadStartFlag.put(init,Boolean.FALSE);
        }

        // 初始化内存队列
        MessageBlockQueue.initMessageQueue(nettyProperties.getMessageMapCapacity(),nettyProperties.getMessageQueueCapatity());

        // http下，target需要动态获取
        if(this.nettyProperties.getStartUpMode().equals(StartUpMode.HTTP)){
            this.target = null;
        }

    }

    @Override
    public void destroy() throws Exception {

        while (MessageBlockQueue.getSize()!=0){
           logger.info("{} : wait for all message consumered .....",this.getClass().getName());
            Thread.sleep(1000);
        }
        logger.info("{} : already message consumered .....",this.getClass().getName());
        MessageBlockQueue.shutdown = Boolean.TRUE;
    }
}
